package com.jida.hadoop.mr.dataclean;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.jida.hadoop.mr.tools.TMatcher;
import com.jida.hadoop.mr.tools.TProperties;

import org.apache.hadoop.mapreduce.Mapper;

public class DxFileClear {

	/**
	 * @param 
	 * args[0]：原始数据输入路径
	 * args[1]：结果输出数据
	 * Description：通过map处理，将不符合规范、特殊格式（配置文件、图片等）数据过滤掉。
	 * */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		//设置map内存使用
		conf.set("mapreduce.map.memory.mb", "2048");
		Job job = new Job(conf, "Dx_FileClear");
		job.setJarByClass(DxFileClear.class);
		//map类
		job.setMapperClass(ClearMapper.class);
		//map输出k-v数据类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		//不使用reduce
		job.setNumReduceTasks(0);
		//结果数据输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setInputFormatClass(TextInputFormat.class);
		//将输入文件切片最小64M，最大128M
	        TextInputFormat.setMinInputSplitSize(job,1024*1024*128L); // 设置最小分片大小
	        TextInputFormat.setMaxInputSplitSize(job,1024*1024*128L); // 设置最大分片大小
	        //输入路径
		FileInputFormat.addInputPath(job, new Path("file:///D:/测试数据/dx/原始数据/*"));
		//输出路径
		FileOutputFormat.setOutputPath(job, new Path("file:///D:/测试数据/dx/01.DxFileClear"));
		//提交作业 判断退出条件（0正常退出，1非正常退出）
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
	
	public static class ClearMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
		//定义输出类型
		private Text ovalue = new Text();
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] str = value.toString().split( TProperties.getValue("filesplit"));
			// 判断数据字段数量，判断url字段是否有效，去除特定格式
			if ((str.length + "").equals(TProperties.getValue("filelength"))
					&& !"http://".equals(str[14])
					&& !"https://".equals(str[14])
					&& !"".equals(str[14])
					&& !(str[14].toLowerCase()).matches(TProperties.getValue("fileclear"))) {

				StringBuffer sb = new StringBuffer();
				if (!str[14].startsWith("http://") && !str[14].startsWith("https://")) {
					str[14] = "http://" + str[14];
				}
				// 获取域名
				String domain = str[14].split("/", -1)[2];
				// 去除端口
				if (domain.indexOf(":") >= 0) {
					domain = domain.split("\\:", -1)[0];
				}
				// 用户号码|一级域名|URL地址
				sb.append(str[1]).append(TProperties.getValue("outfilesplit"))
						.append(TMatcher.getDomain(domain)).append(TProperties.getValue("outfilesplit"))
						.append(str[14]);

				ovalue = new Text(sb.toString());
				context.write(NullWritable.get(), ovalue);
			}
		}
	}
}